package com.dingwen.treasure.core.component;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import com.dingwen.treasure.base.constant.TaskConstant;
import com.dingwen.treasure.base.enums.TaskState;
import com.dingwen.treasure.base.enums.TaskType;
import com.dingwen.treasure.common.redis.service.RedisService;
import com.dingwen.treasure.core.exception.ProgressBarException;
import com.dingwen.treasure.core.vo.TaskVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.cache.RedisCache;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * 进度条任务 b</br>
 * <p> 范型说明：</p>
 * <ul>
 *     <li> ReturnValue: 任务执行返回值</li>
 *     <li> Param: 组装任务所需参数</li>
 * </ul>
 *
 * @author dingwen
 * @date 2022/11/07
 */
@Slf4j
public abstract class AbstractProgressBarTask<ReturnValue, Param> {

    /**
     * 任务执行
     *
     * @param taskId    任务id
     * @param taskType  任务类型
     * @param taskTotal 任务总数量
     * @return 任务进度对象
     */
    public TaskVo execute(String taskId,
                          TaskType taskType,
                          Integer taskTotal,
                          Param param) {

        // 任务执行前检查
        TaskVo taskVo = executePreInspection(taskId, taskType, taskTotal);
        String taskKey = taskVo.getTaskKey();

        // 任务正在进行中
        if (!TaskState.NONE.getCode().equals(taskVo.getTaskStatus())) {
            return taskVo;
        }

        // 开始提交任务
        try {
            List<Supplier<ReturnValue>> tasks = getTasks(param);
            Assert.isTrue(tasks.size() > 0, "待执行任务不能为零");

            // 提交任务
            submitAsyncTask(tasks, taskKey, taskId, taskType, param);
            return taskVo;

        } catch (Exception e) {
            String msg = StrUtil.format("{},进度条任务执行发生异常,终止：{}", taskKey, e.getMessage());
            updateTaskProcessForFinalFail(taskKey, msg);
            log.error(msg);
            throw new ProgressBarException(msg);
        }

    }

    /**
     * 查询任务进度
     *
     * @param taskId   任务id
     * @param taskType 任务类型
     * @return {@link TaskVo}
     */
    public TaskVo queryProcess(String taskId, TaskType taskType) {
        String taskKey = generateTaskKey(taskId, taskType);
        TaskVo taskVo = getCacheTaskVo(taskKey);

        if (taskVo.getTaskStatus().equals(TaskState.SUCCESS.getCode())
                || taskVo.getTaskStatus().equals(TaskState.FAIL.getCode())) {
            removeTaskVo(taskKey);
        }
        return taskVo;
    }

    /**
     * 任务执行前置检查
     *
     * @param taskId    任务id
     * @param taskType  任务类型
     * @param taskTotal 任务总数量
     * @return 任务进度对象
     */
    protected TaskVo executePreInspection(String taskId, TaskType taskType, Integer taskTotal) {
        RedisService redisService = getRedisService();
        // 任务key
        String taskKey = generateTaskKey(taskId, taskType);
        // 任务进度视图
        TaskVo taskVo = buildTaskVo(taskTotal, taskKey);
        if (!checkTaskKeyIfAbsent(taskKey)) {
            // 任务已存在
            return taskVo.setMsg(TaskConstant.TASK_MSG_ING)
                    .setTaskStatus(TaskState.ING.getCode());
        }
        redisService.deleteObject(taskKey);
        redisService.setCacheMap(taskKey, BeanUtil.beanToMap(taskVo));

        return taskVo;
    }

    /**
     * 生成任务key
     *
     * @param taskId   任务id
     * @param taskType 任务类型
     * @return {@link String}
     */
    protected String generateTaskKey(String taskId, TaskType taskType) {
        return TaskConstant.TASK_KEY_PREFIX.concat(taskType.getCode()).concat(":").concat(StrUtil.toString(taskId));
    }

    /**
     * 构建任务视图
     *
     * @param taskTotal 总任务数
     * @param taskKey   任务key
     * @return {@link TaskVo}
     */
    protected TaskVo buildTaskVo(Integer taskTotal, String taskKey) {
        return TaskVo.builder()
                .taskStatus(TaskState.NONE.getCode())
                .msg(TaskConstant.TASK_MSG_INIT)
                .totalNum(taskTotal)
                .sucNum(0)
                .failNum(0)
                .taskKey(taskKey)
                .build();
    }

    /**
     * 检查当前任务key是否存在
     *
     * @param taskKey 任务key
     * @return boolean
     */
    protected boolean checkTaskKeyIfAbsent(String taskKey) {
        RedisService redisService = getRedisService();
        return redisService.setIfAbsent(taskKey, "", 2L, TimeUnit.MINUTES);
    }


    /**
     * 更新任务进度
     *
     * @param taskKey  任务键
     * @param fieldKey 域键
     * @param value    值
     */
    protected void updateTaskProgress(String taskKey, String fieldKey, String value) {
        getRedisService().setCacheMapValue(taskKey, fieldKey, value);
    }

    /**
     * 更新失败任务进度
     *
     * @param taskKey 任务键
     * @param msg     失败消息
     */
    protected void updateTaskProcessForFail(String taskKey, String msg) {
        RedisService redisService = getRedisService();
        redisService.incrementForHashByKey(taskKey, TaskVo.FAIL_NUM, 1L);
        updateTaskProcessForFinalFail(taskKey, msg);
    }

    protected void updateTaskProcessForFinalFail(String taskKey, String msg) {
        updateTaskProgress(taskKey, TaskVo.TASK_STATUS, TaskState.FAIL.getCode());
        updateTaskProgress(taskKey, TaskVo.MSG, msg);
    }

    /**
     * 更新成功任务进度
     *
     * @param taskKey 任务键
     */
    protected void updateTaskProcessForSuccess(String taskKey, String taskStatus) {
        getRedisService().incrementForHashByKey(taskKey, TaskVo.SUC_NUM, 1L);
        updateTaskProcessForFinalSuccess(taskKey, TaskConstant.TASK_MSG_ING, taskStatus);
    }

    /**
     * 更新任务过程最终成功
     *
     * @param taskKey    任务关键
     * @param msg        味精
     * @param taskStatus 任务状态
     */
    protected void updateTaskProcessForFinalSuccess(String taskKey, String msg, String taskStatus) {
        updateTaskProgress(taskKey, TaskVo.TASK_STATUS, taskStatus);
        updateTaskProgress(taskKey, TaskVo.MSG, msg);
    }


    /**
     * 获取缓存任务执行信息
     *
     * @param taskKey 任务关键
     * @return {@link TaskVo}
     */
    protected TaskVo getCacheTaskVo(String taskKey) {
        TaskVo taskVo = BeanUtil.toBean(getRedisService().getCacheMap(taskKey), TaskVo.class);
        Assert.notNull(taskVo.getTaskKey(), StrUtil.format("{}任务未在执行中", taskKey));
        return taskVo;
    }


    /**
     * 删除任务任务执行信息
     *
     * @param taskKey 任务关键
     */
    protected void removeTaskVo(String taskKey) {
        getRedisService().deleteObject(taskKey);
    }

    /**
     * 提交异步任务
     *
     * @param tasks    任务
     * @param taskKey  任务key
     * @param taskId   任务id
     * @param taskType 任务类型
     * @param param    参数
     */
    private void submitAsyncTask(List<Supplier<ReturnValue>> tasks,
                                 String taskKey,
                                 String taskId,
                                 TaskType taskType,
                                 Param param) {
        // 获取执行任务线程池
        ThreadPoolTaskExecutor asyncExecutor = getAsyncExecutor();
        asyncExecutor.execute(() -> {
            List<CompletableFuture> futures = new ArrayList<>(tasks.size());
            List<ReturnValue> resultData = new ArrayList<>(tasks.size());
            tasks.forEach(task -> {
                        CompletableFuture<ReturnValue> future = CompletableFuture.supplyAsync(task, asyncExecutor)

                                // 当任务完成时
                                .whenComplete((re, ex) -> {
                                    // 执行进度更新
                                    resultData.add(re);
                                    updateTaskProcessForSuccess(taskKey, TaskState.NONE.getCode());
                                })

                                // 当任务执行发生异常时
                                .exceptionally((ex) -> {
                                    log.error("任务:{}执行异常:{}", taskKey, ex.getMessage());
                                    updateTaskProcessForFail(taskKey, ex.getMessage());
                                    return null;
                                });
                        futures.add(future);
                    }
            );
            // 所有任务已完成时
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{})).join();
            log.info("任务:{},总数:{},已完成", taskKey, tasks.size());

            // 执行后置处理操作
            try {
                after(taskId, taskType, taskKey, param, resultData);
            } catch (Exception e) {
                String msg = StrUtil.format("{}进度条任务执行后置处理发生异常", taskKey);
                updateTaskProcessForFinalFail(taskKey, msg);
                log.error(msg);
                throw new ProgressBarException(msg);
            }

            // 最终完成
            updateTaskProcessForFinalSuccess(taskKey, TaskConstant.TASK_MSG_FINISHED, TaskState.SUCCESS.getCode());

        });
    }

    /**
     * 获取缓存操作类 </br>
     * <p>不能为空，否则直接抛出异常</p>
     *
     * @return {@link RedisCache}
     */
    protected abstract RedisService getRedisService();


    /**
     * 得到线程池 </br>
     * <p>不能为空，否则直接抛出异常</p>
     *
     * @return {@link ThreadPoolTaskExecutor}
     */
    protected abstract ThreadPoolTaskExecutor getAsyncExecutor();

    /**
     * 得到需要执行的任务
     *
     * @param param 组装执行任务所需参数
     * @return {@link List}<{@link Supplier}<{@link ReturnValue}>>
     * @throws ProgressBarException 异常
     */
    protected abstract List<Supplier<ReturnValue>> getTasks(Param param);


    /**
     * 收尾方法
     *
     * @param taskId     任务id
     * @param taskType   {@link TaskType} 任务类型
     * @param taskKey    任务key
     * @param param      组装执行任务所需参数
     * @param resultData 任务执行结果数据
     * @throws ProgressBarException 进度条例外
     */
    protected abstract void after(
            String taskId,
            TaskType taskType,
            String taskKey,
            Param param,
            List<ReturnValue> resultData
    );

}
